package com.baihejia.common.dubbo.server.libs;

import com.baihejia.common.dubbo.server.Configuration;
import redis.clients.jedis.*;

import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.util.*;

public class RedisClient {

    private static ShardedJedisPool shardedJedisPool = null;

    static {
        int timeout = 50000;
        int maxActive = 500;
        int maxIdle = 50;

        List<JedisShardInfo> shards = new ArrayList<>();

        RedisConnectionInfo redisConnInfo = new RedisConnectionInfo(Configuration.get("redis"), timeout);
        shards.add(redisConnInfo.getShardInfo());

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(maxActive);
        poolConfig.setMaxIdle(maxIdle);
        shardedJedisPool =  new ShardedJedisPool(poolConfig, shards, ShardedJedis.DEFAULT_KEY_TAG_PATTERN);
    }

    private static class RedisConnectionInfo {
        private final String host;
        private final int port;
        private final String password;
        private final int timeout;

        RedisConnectionInfo(String redisUrl, int timeout) {
            URI redisUri;
            try {
                redisUri = new URI(redisUrl);
            } catch (URISyntaxException e) {
                throw new RuntimeException("Bad configuration for redis: unable to parse redis url (" + redisUrl + ")");
            }

            host = redisUri.getHost();

            if (redisUri.getPort() > 0) {
                port = redisUri.getPort();
            } else {
                port = Protocol.DEFAULT_PORT;
            }

            String userInfo = redisUri.getUserInfo();
            if (userInfo != null) {
                String[] parsedUserInfo = userInfo.split("::");
                password = parsedUserInfo[parsedUserInfo.length - 1];
            } else {
                password = null;
            }

            this.timeout = timeout;
        }

        JedisPool getConnectionPool() {
            if (password == null) {
                return new JedisPool(new JedisPoolConfig(), host, port, timeout);
            }

            return new JedisPool(new JedisPoolConfig(), host, port, timeout, password);
        }

        JedisShardInfo getShardInfo() {
            JedisShardInfo si = new JedisShardInfo(host, port, timeout);
            si.setPassword(password);
            return si;
        }
    }

    public static String setex(String cacheKey, int seconds, String value){
        return execute(cacheKey, jedis -> jedis.setex(cacheKey, seconds, value));
    }

    public static String set(String cacheKey, String value){
        return execute(cacheKey, jedis -> jedis.set(cacheKey, value));
    }

    public static String get(String cacheKey){
        return execute(cacheKey, jedis -> jedis.get(cacheKey));
    }

    public static long incr(String cacheKey){
        return execute(cacheKey, jedis -> jedis.incr(cacheKey));
    }

    public static long incrBy(String cacheKey, long step){
        return execute(cacheKey, jedis -> jedis.incrBy(cacheKey, step));
    }

    public static long decr(String cacheKey){
        return execute(cacheKey, jedis -> jedis.decr(cacheKey));
    }

    public static boolean exists(String cacheKey){
        return execute(cacheKey, jedis -> jedis.exists(cacheKey));
    }

    public static List<String> lrange(String cacheKey, long start, long end){
        return execute(cacheKey, jedis -> jedis.lrange(cacheKey, start, end));
    }

    public static long lpush(String cacheKey, String...strings){
        return execute(cacheKey, jedis -> jedis.lpush(cacheKey, strings));
    }

    public static long del(String cacheKey){
        return execute(cacheKey, jedis -> jedis.del(cacheKey));
    }

    public static long del(String... keys){
        return execute(keys.toString(), jedis -> jedis.del(keys));
    }

    public static Map<String, String> hgetAll(String cacheKey){
        return execute(cacheKey, jedis -> jedis.hgetAll(cacheKey));
    }

    public static String hget(String cacheKey, String field){
        return execute(cacheKey, jedis -> jedis.hget(cacheKey, field));
    }

    public static List<String> hmget(String cacheKey, String...fields){
        return execute(cacheKey, jedis -> jedis.hmget(cacheKey, fields));
    }

    public static Long hincrBy(String cacheKey, String field, long value){
        return execute(cacheKey, jedis -> jedis.hincrBy(cacheKey, field, value));
    }
    public static long hset(String cacheKey, String field, String value){
        return execute(cacheKey, jedis -> jedis.hset(cacheKey, field, value));
    }

    public static long hdel(String cacheKey, String...fields){
        return execute(cacheKey, jedis -> jedis.hdel(cacheKey, fields));
    }

    public static boolean sismember(String cacheKey, String field){
        return execute(cacheKey, jedis -> jedis.sismember(cacheKey, field));
    }

    public static Set<String> zrange(String cacheKey, long start, long end){
        return execute(cacheKey, jedis -> jedis.zrange(cacheKey, start, end));
    }

    public static Set<String> zrevrange(String cacheKey, long start, long end){
        return execute(cacheKey, jedis -> jedis.zrevrange(cacheKey, start, end));
    }

    public static Set<String> zrangeByScore(String cacheKey, double min, double max){
        return execute(cacheKey, jedis -> jedis.zrangeByScore(cacheKey, min, max));
    }

    public static Set<String> zrangeByScore(final String cacheKey, final double min, final double max, final int page, final int pageSize){
        return execute(cacheKey, jedis -> jedis.zrangeByScore(cacheKey, min, max, (page - 1) * pageSize, pageSize));
    }

    public static Set<String> zrevrangeByScore(String cacheKey, double min, double max){
        return execute(cacheKey, jedis -> jedis.zrevrangeByScore(cacheKey, max, min));
    }

    public static Set<String> zrevrangeByScore(final String cacheKey, final double min, final double max, final int page, final int pageSize) {
        return execute(cacheKey, jedis -> jedis.zrevrangeByScore(cacheKey, max, min, (page - 1) * pageSize, pageSize));
    }

    public static long zadd(String cacheKey, double scores, String member){
        return execute(cacheKey, jedis -> jedis.zadd(cacheKey, scores, member));
    }

    public static long zadd(String cacheKey, double score, String... members){
        Map<String, Double> scoreMembers = new HashMap<>();
        for (String member : members) {
            scoreMembers.put(member, score);
        }
        return execute(cacheKey, jedis -> jedis.zadd(cacheKey, scoreMembers));
    }

    public static long zcard(String cacheKey){
        return execute(cacheKey, jedis -> jedis.zcard(cacheKey));
    }

    public static Long zrank(String cacheKey, String member){
        return execute(cacheKey, jedis -> jedis.zrank(cacheKey, member));
    }

    public static long zrem(String cacheKey, String...members){
        return execute(cacheKey, jedis -> jedis.zrem(cacheKey, members));
    }

    public static Double zscore(String cacheKey, String member){
        return execute(cacheKey, jedis -> jedis.zscore(cacheKey, member));
    }

    public static Double zincrby(String cacheKey, double score, String member){
        return execute(cacheKey, jedis -> jedis.zincrby(cacheKey, score, member));
    }

    public static long zcount(String cacheKey, double min, double max){
        return execute(cacheKey, jedis -> jedis.zcount(cacheKey, min, max));
    }

    public static Set<Tuple> zrevrangeWithScores(String cacheKey, long start, long end){
        return execute(cacheKey, jedis -> jedis.zrevrangeWithScores(cacheKey, start, end));
    }

    public static Set<Tuple> zrangeWithScores(String cacheKey, long start, long end){
        return execute(cacheKey, jedis -> jedis.zrangeWithScores(cacheKey, start, end));
    }

    public static long expire(String cacheKey, int seconds){
        return execute(cacheKey, jedis -> jedis.expire(cacheKey, seconds));
    }

    public static long expireAt(String cacheKey, long unixTime){
        return execute(cacheKey, jedis -> jedis.expireAt(cacheKey, unixTime));
    }

    public static Set<String> smembers(String cacheKey){
        return execute(cacheKey, jedis -> jedis.smembers(cacheKey));
    }

    public static long sadd(String cacheKey, String...members){
        return execute(cacheKey, jedis -> jedis.sadd(cacheKey, members));
    }

    public static boolean smembers(String cacheKey, String member){
        return execute(cacheKey, jedis -> jedis.sismember(cacheKey, member));
    }

    public static String srandmember(String cacheKey) {
        return execute(cacheKey, jedis -> jedis.srandmember(cacheKey));
    }

    public static Long zunionstore(final String cacheKey, final String... otherKeys) {
        return execute(cacheKey, jedis -> {
            ZParams zParams = new ZParams();
            zParams.weightsByDouble(1);
            zParams.aggregate(ZParams.Aggregate.MIN);
            return jedis.zunionstore(cacheKey, zParams, otherKeys);
        });
    }

    public static long srem(String cacheKey, String...members){
        return execute(cacheKey, jedis -> jedis.srem(cacheKey, members));
    }

    public static Set<String> keys(String keyPattern){
        try(ShardedJedis shardedJedis = shardedJedisPool.getResource()){
            Collection<Jedis> jedises = shardedJedis.getAllShards();
            Set<String> result = new HashSet<>();
            for (Jedis jedis : jedises){
                try {
                    result.addAll(jedis.keys(keyPattern));
                }finally {
                    jedis.close();
                }
            }
            return result;
        }
    }

    public static List<String> mget(String...keys){
        if (keys == null || keys.length == 0){
            return Collections.emptyList();
        }
        try(ShardedJedis shardedJedis = shardedJedisPool.getResource()){
            Collection<Jedis> jedises = shardedJedis.getAllShards();
            List<String> resultT = new ArrayList<>(keys.length);
            for (String ignored : keys) {
                resultT.add(null);
            }
            for (Jedis jedis : jedises){
                try {
                    List<String> cacheValues = jedis.mget(keys);
                    for (int i = 0; i < keys.length; i++){
                        if (resultT.get(i) == null && cacheValues.get(i) != null){
                            resultT.set(i, cacheValues.get(i));
                        }
                    }
                }finally {
                    jedis.close();
                }
            }
            return resultT;
        }
    }

    private static <T> T execute(byte[] key, RedisCallback<T> callback) {
        ShardedJedis shardedJedis = shardedJedisPool.getResource();

        try(Jedis jedis = shardedJedis.getShard(key)){
            return callback.execute(jedis);
        }finally {
            shardedJedis.close();
        }
    }

    private static <T> T execute(String key, RedisCallback<T> callback){
        return execute(key.getBytes(Charset.forName("utf-8")), callback);
    }


    @FunctionalInterface
    public interface RedisCallback<T> {
        T execute(Jedis jedis);
    }

}
